package javaVersion.learn.stateProcess.broadCastStateDemo;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Map;

/**
 * @program: myFlink
 * @description: 自定义广播流处理方法
 * @author: WincoMa
 * @create: 2020-07-28 09:41
 **/
public class CustomeBroadcastProcessFunction extends
        BroadcastProcessFunction<
                //输入1：原始数据
                Tuple4<String,String,String,Integer>,
                //输入2：具体规则（mysql广播数据）
                Map<String,
                        Map<String,Integer>>,
                //输出：根据业务需求处理的输出数据
                Tuple6<String,String,String,Integer,String,Integer>
                > {
    //定义广播状态描述符
    MapStateDescriptor<Void, Map<String, Map<String, Integer>>> configDescriptor = new MapStateDescriptor<>(
            "config",
//            BasicTypeInfo.VOID_TYPE_INFO,
            Types.VOID,
            TypeInformation.of(new TypeHint<Map<String, Map<String, Integer>>>() {
            })
    );



    /**
     * 具体广播数据格式化处理逻辑
     * @param value 原始数据
     * @param ctx 上下文
     * @param out 返回值
     * @throws Exception
     */
    @Override
    public void processElement(Tuple4<String, String, String, Integer> value, ReadOnlyContext ctx, Collector<Tuple6<String, String, String, Integer, String, Integer>> out) throws Exception {
        //获取用户id
        String userID = value.f0;
        //读取广播（状态）
        ReadOnlyBroadcastState<Void, Map<String, Map<String, Integer>>> broadcastState = ctx.getBroadcastState(configDescriptor);
        //获取用户列表
        Map<String, Map<String, Integer>> broadcastStateUserInfo = broadcastState.get(null);
        Map<String, Integer> userInfo = broadcastStateUserInfo.getOrDefault(userID, null);
        //获取userInfo的key和value
        for (Map.Entry<String, Integer> entry : userInfo.entrySet()) {
            if (userInfo != null) {
                //如果查到了用户信息，则标准化输出
                out.collect(Tuple6.of(value.f0,value.f1,value.f2,value.f3,entry.getKey(),entry.getValue()));
            }
        }
    }

    /**
     * 处理广播数据，并更新状态
     * @param value 广播数据
     * @param ctx   上下文
     * @param out   返回值
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(Map<String, Map<String, Integer>> value, Context ctx, Collector<Tuple6<String, String, String, Integer, String, Integer>> out) throws Exception {
        BroadcastState<Void, Map<String, Map<String, Integer>>> broadcastState = ctx.getBroadcastState(configDescriptor);
        broadcastState.clear();
        broadcastState.put(null,value);
    }
}
